package com.atguigu.flink.cep;

import com.atguigu.flink.pojo.LoginEvent;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.util.List;
import java.util.Map;

/**
 * Created by Smexy on 2022/11/26
 *
 * 如果同一用户（可以是不同IP）在2秒之内连续两次登录，都失败，就认为存在恶意登录的风险
 */
public class Demo12_CEP_Login
{
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        WatermarkStrategy<LoginEvent> watermarkStrategy = WatermarkStrategy
            .<LoginEvent>forMonotonousTimestamps()
             .withTimestampAssigner((e, r) -> e.getEventTime() * 1000);

        KeyedStream<LoginEvent, Long> ds = env.readTextFile("data/LoginLog.csv")
                                         .map(new MapFunction<String, LoginEvent>()
                                         {
                                             @Override
                                             public LoginEvent map(String value) throws Exception {
                                                 String[] words = value.split(",");
                                                 return new LoginEvent(
                                                     Long.valueOf(words[0]),
                                                     words[1],
                                                     words[2],
                                                     Long.valueOf(words[3])
                                                 );
                                             }
                                         })
                                         .assignTimestampsAndWatermarks(watermarkStrategy)
                                              //把同一用户的数据放到一起
                                         .keyBy(LoginEvent::getUserId);

        //定义一个Pattern  2秒之内连续两次登录，都失败
        Pattern<LoginEvent, LoginEvent> pattern = Pattern.<LoginEvent>begin("fail")
            .where(new SimpleCondition<LoginEvent>()
            {
                @Override
                public boolean filter(LoginEvent value) throws Exception {
                    return "fail".equals(value.getEventType());
                }
            })
            .times(2) //两次失败
            .consecutive() // 严格连续，紧挨着
            .within(Time.seconds(2));//< 2s


        //作用到流上
        CEP.pattern(ds,pattern)
           .select(new PatternSelectFunction<LoginEvent, String>()
           {
               // 匹配到的数据
               @Override
               public String select(Map<String, List<LoginEvent>> pattern) throws Exception {
                   Long userId = pattern.get("fail").get(0).getUserId();
                   return userId + "恶意登录...";
               }
           })
           .print();

        env.execute();


    }
}
